package com.quanyan.ic.video.websocket;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.quanyan.ic.video.websocket.transfer.MessageType;
import com.quanyan.ic.video.websocket.transfer.WebSocketMessage;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by zhuo.shi on 2017/8/22.
 */
@ChannelHandler.Sharable
public class WebSocketServerHandler extends ChannelInboundHandlerAdapter {

    private static final Logger logger = LoggerFactory.getLogger(WebSocketServerHandler.class);

    private Map<String, Channel> sessionCache = new ConcurrentHashMap<>();

    private WebSocketServerHandshaker handshaker;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof FullHttpRequest) {
            handleHttpRequest(ctx, (FullHttpRequest) msg);
        } else if (msg instanceof WebSocketFrame) {
            handleWebSocketFrame(ctx, (WebSocketFrame) msg);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
        sessionCache.values().remove(ctx.channel());
        logger.warn("catch exception: {}, remove channel {}", cause.getMessage(), ctx.channel());
        logger.info("sessionCache={}", sessionCache);
    }

    private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request) {
        if (!request.getDecoderResult().isSuccess() || !"websocket".equals(request.headers().get("Upgrade"))) {
            FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST);
            ByteBuf byteBuf = Unpooled.copiedBuffer(response.getStatus().toString(), CharsetUtil.UTF_8);
            response.content().writeBytes(byteBuf);
            byteBuf.release();
            response.headers().set("Content-Length", response.content().readableBytes());
            ChannelFuture f = ctx.writeAndFlush(response);
            f.addListener(ChannelFutureListener.CLOSE);
            return;
        }

        WebSocketServerHandshakerFactory handshakerFactory = new WebSocketServerHandshakerFactory("ws://localhost:9263/alarm", null, false);
        handshaker = handshakerFactory.newHandshaker(request);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        } else {
            handshaker.handshake(ctx.channel(), request);
        }
    }

    private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        if (!(frame instanceof TextWebSocketFrame)) {
            throw new UnsupportedOperationException("frame type " + frame.getClass().getName() + " not supported");
        }
        String requestStr = ((TextWebSocketFrame) frame).text();
        logger.info("received websocket request: {}", requestStr);

        WebSocketMessage<Map<String, Object>> message = JSON.parseObject(requestStr, new TypeReference<WebSocketMessage<Map<String, Object>>>(){});
        if (MessageType.HANDSHAKE.value().equals(message.getType().value())){
            String sessionId = message.getData().get("sessionId").toString();
            sessionCache.put(sessionId, ctx.channel());
            logger.info("sessionCache=" + sessionCache.toString());
            WebSocketMessage<String> respMsg = new WebSocketMessage<>();
            respMsg.setType(MessageType.HANDSHAKE);
            respMsg.setData("client [" + sessionId + "] handshake success");
            ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(respMsg)));
        }
    }

    public Map<String, Channel> getSessionCache() {
        return sessionCache;
    }

    public void setSessionCache(Map<String, Channel> sessionCache) {
        this.sessionCache = sessionCache;
    }
}
